package org.example.iotdb.flink.process;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import org.apache.commons.compress.utils.Lists;
import org.apache.iotdb.flink.Event;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @Auther: kevin
 * @Description:
 * @Company: 上海博般数据技术有限公司
 * @Version: 1.0.0
 * @Date: Created in 13:23 2021/8/20
 * @ProjectName: iotdb-learn
 */
public class CollectEventProcessWindow extends ProcessWindowFunction<
        Tuple2<String, Event>,
        List<Event>,
        Tuple,
        TimeWindow> {

    @Override
    public void process(Tuple tuple, Context context, Iterable<Tuple2<String, Event>> iterable, Collector<List<Event>> collector) throws Exception {
        List<Tuple2<String, Event>> list = Lists.newArrayList(iterable.iterator());
        List<Event> reslt = list.stream()
                .map(t -> t.f1)
                .collect(Collectors.toList());
        collector.collect(reslt);
    }

}
